package com.edata.bigdata.streaming.kafka

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}

import java.util.Properties

class ProducerSinks[K, V](createProducer: () => KafkaProducer[K, V]) extends Serializable {

  lazy val producer = createProducer()

  def send(topic: String, key: K, value: V) = {
    producer.send(new ProducerRecord[K, V](topic, key, value))
  }

  def send(topic: String, value: V) = {
    producer.send(new ProducerRecord[K, V](topic, value))
  }

}


object ProducerSinks {
  def apply[K, V](conf: Properties): ProducerSinks[K, V] = {
    var func = () => {
      val prod = new KafkaProducer[K, V](conf)
      //addshutdownHook 一般会在 JVM 结束前运行，一般用于一些清理工作
      //此处负责关闭 KafkaProducer。
      sys.addShutdownHook({
        prod.close()
      })
      prod
    }
    new ProducerSinks[K, V](func)
  }

}
